package com.xxx;

import com.google.protobuf.ProtocolStringList;
import io.grpc.stub.StreamObserver;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class HelloServiceImpl extends HelloServiceGrpc.HelloServiceImplBase {
    /**
     * 1、接收client提交的参数 接收单个参数
     * 2、业务处理 比如 service+dao 调用对应的业务功能
     * 3、提供返回值
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void hello(HelloProto.HelloRequest request, StreamObserver<HelloProto.HelloResponse> responseObserver) {
        // 1、接收client的请求参数
        String name = request.getName();
        // 2、业务处理 这里用输出请求参数到控制台代替业务处理
        System.out.println("name: " + name);
        // 3、封装响应
        // 3.1创建响应对象的构建者
        HelloProto.HelloResponse.Builder builder = HelloProto.HelloResponse.newBuilder();
        // 3.2填充数据
        builder.setResult("hello server response ok");
        // 3.3封装响应
        HelloProto.HelloResponse helloResponse = builder.build();

        // 响应信息回传给client
        responseObserver.onNext(helloResponse);
        // 通知client 整个服务结束 底层返回标记 client监听标记是grpc做的
        responseObserver.onCompleted();
    }

    /**
     * 接收多个参数
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void hello1(HelloProto.HelloRequest1 request, StreamObserver<HelloProto.HelloResponse1> responseObserver) {
        ProtocolStringList nameList = request.getNameList();
        for (String name : nameList) {
            System.out.println(name);
        }
        System.out.println("HelloServiceImpl.hello1");

        HelloProto.HelloResponse1.Builder builder = HelloProto.HelloResponse1.newBuilder();
        builder.setResult("hello1 response ok");
        HelloProto.HelloResponse1 helloResponse1 = builder.build();

        responseObserver.onNext(helloResponse1);
        responseObserver.onCompleted();
    }

    /**
     * 服务端流式响应
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void c2ss(HelloProto.HelloRequest request, StreamObserver<HelloProto.HelloResponse> responseObserver) {
        // 接收客户端的请求参数
        String name = request.getName();
        // 业务处理
        System.out.println(name);
        // 根据业务的处理结果提供响应
        for (int i = 0; i < 10; i++) {
            HelloProto.HelloResponse.Builder builder = HelloProto.HelloResponse.newBuilder();
            builder.setResult("server response stream " + i);
            HelloProto.HelloResponse helloResponse = builder.build();
            responseObserver.onNext(helloResponse);
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        responseObserver.onCompleted();
    }

    /**
     * 客户端流式RPC
     * 比较特别的是返回的是一个StreamObserver
     * 由于该接口的返回不是流式，如果在StreamObserver<HelloProto.HelloRequest>()的onNext()方法去写返回的话会报警告：
     * 警告: Cancelling the stream with status Status{code=INTERNAL, description=Too many responses, cause=null}
     * 因此这种情况只能在客户端的请求流结束onCompleted()方法中去进行响应返回
     *
     * @param responseObserver 监听client端的请求
     * @return
     */
    @Override
    public StreamObserver<HelloProto.HelloRequest> cs2s(StreamObserver<HelloProto.HelloResponse> responseObserver) {

        return new StreamObserver<HelloProto.HelloRequest>() {
            private List<HelloProto.HelloRequest> requestList = new ArrayList<>();

            @Override
            public void onNext(HelloProto.HelloRequest helloRequest) {
                requestList.add(helloRequest);
                // 对client端每次请求进行处理
                System.out.println("接收到了client发送的一条消息: " + helloRequest.getName());
                HelloProto.HelloResponse.Builder builder = HelloProto.HelloResponse.newBuilder();
                builder.setResult("cs2s 接收到client的第 " + requestList.size() + " 条消息");
                HelloProto.HelloResponse helloResponse = builder.build();
                // 这样会被警告，并出现错误的结果 Cancelling the stream with status Status{code=INTERNAL, description=Too many responses, cause=null}
                // responseObserver.onNext(helloResponse);
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onCompleted() {
                // client端所有请求消息发送完成后 统一处理所有的请求数据
                System.out.println("client 的所有消息接收完成，服务端统一处理: " + requestList.size());
                HelloProto.HelloResponse.Builder builder = HelloProto.HelloResponse.newBuilder();
                builder.setResult("cs2s 接收到client共 " + requestList.size() + " 条消息");
                HelloProto.HelloResponse helloResponse = builder.build();

                responseObserver.onNext(helloResponse);

                responseObserver.onCompleted();
            }
        };
    }

    /**
     * cliet server 双向流式RPC
     *
     * @param responseObserver
     * @return
     */
    @Override
    public StreamObserver<HelloProto.HelloRequest> cs2ss(StreamObserver<HelloProto.HelloResponse> responseObserver) {
        return new StreamObserver<HelloProto.HelloRequest>() {
            private List<HelloProto.HelloRequest> requestList = new ArrayList<>();

            @Override
            public void onNext(HelloProto.HelloRequest helloRequest) {
                requestList.add(helloRequest);
                // 对client端每次请求进行处理
                System.out.println("接收到了client发送的一条消息: " + helloRequest.getName());
                HelloProto.HelloResponse.Builder builder = HelloProto.HelloResponse.newBuilder();
                builder.setResult("cs2s 接收到client的第 " + requestList.size() + " 条消息");
                HelloProto.HelloResponse helloResponse = builder.build();
                // 双向流不会出现错误的结果 Cancelling the stream with status Status{code=INTERNAL, description=Too many responses, cause=null}
                responseObserver.onNext(helloResponse);
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onCompleted() {
                // client端所有请求消息发送完成后 统一处理所有的请求数据
                System.out.println("client 的所有消息接收完成，服务端统一处理: " + requestList.size());
                HelloProto.HelloResponse.Builder builder = HelloProto.HelloResponse.newBuilder();
                builder.setResult("cs2s 接收到client共 " + requestList.size() + " 条消息");
                HelloProto.HelloResponse helloResponse = builder.build();

                responseObserver.onNext(helloResponse);

                responseObserver.onCompleted();
            }
        };
    }
}
